namespace Crux.Core.Websockets.Redis.Metadata;

/// <summary>
///
/// </summary>
internal class AckHandler : IDisposableObservable
{
    /// <summary>
    ///
    /// </summary>
    private readonly ConcurrentDictionary<int, AckInfo> _acks = new();
    /// <summary>
    ///
    /// </summary>
    private readonly Timer _timer;
    /// <summary>
    ///
    /// </summary>
    private readonly TimeSpan _ackThreshold = TimeSpan.FromSeconds(30);
    /// <summary>
    ///
    /// </summary>
    private readonly TimeSpan _ackInterval = TimeSpan.FromSeconds(5);
    /// <summary>
    ///
    /// </summary>
    private readonly object _lock = new object();
    public bool IsDisposed { get; private set; }

    /// <summary>
    ///
    /// </summary>
    public AckHandler()
    {
        bool restoreFlow = false;
        try
        {
            if (!ExecutionContext.IsFlowSuppressed())
            {
                ExecutionContext.SuppressFlow();
                restoreFlow = true;
            }
            _timer = new Timer(state => (state as AckHandler)!.CheckAcks(), state: this, dueTime: _ackThreshold, period: _ackInterval);
        }
        finally
        {
            if (restoreFlow)
            {
                ExecutionContext.RestoreFlow();
            }
        }
    }
    /// <summary>
    ///
    /// </summary>
    public Task CreateAck(int id)
    {
        lock (_lock)
        {
            if (IsDisposed)
            {
                return Task.CompletedTask;
            }

            return _acks.GetOrAdd(id, _ => new AckInfo()).Tcs.Task;
        }
    }
    /// <summary>
    ///
    /// </summary>
    public void TriggerAck(int id)
    {
        if (_acks.TryRemove(id, out var ack))
        {
            ack.Tcs.TrySetResult(0);
        }
    }
    /// <summary>
    ///
    /// </summary>
    private void CheckAcks()
    {
        if (IsDisposed) return;

        var utcNow = DateTime.UtcNow;
        foreach (var pair in _acks)
        {
            var elapsed = utcNow - pair.Value.Created;
            if (elapsed > _ackThreshold)
            {
                if (_acks.TryRemove(pair.Key, out var ack))
                {
                    ack.Tcs.TrySetCanceled();
                }
            }
        }
    }
    /// <summary>
    ///
    /// </summary>
    public void Dispose()
    {
        if (this.IsDisposed) return;

        GC.SuppressFinalize(this);
        lock (_lock)
        {
            this.IsDisposed = true;

            _timer.Dispose();
            foreach (var pair in _acks)
            {
                if (_acks.TryRemove(pair.Key, out var ack))
                {
                    ack.Tcs.TrySetCanceled();
                }
            }
        }
    }
    /// <summary>
    ///
    /// </summary>
    private class AckInfo
    {
        /// <summary>
        ///
        /// </summary>
        public TaskCompletionSource<object> Tcs { get; init; }
        /// <summary>
        ///
        /// </summary>
        public DateTime Created { get; init; }
        /// <summary>
        ///
        /// </summary>
        public AckInfo()
        {
            this.Created = DateTime.Now;
            this.Tcs = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
        }
    }
}
